<?php
// +----------------------------------------------------------------------
// | zhanshop-cloud / MqttServEvent.php    [ 2025/1/5 11:22 ]
// +----------------------------------------------------------------------
// | Copyright (c) 2011~2025 zhangqiquan All rights reserved.
// +----------------------------------------------------------------------
// | Author: zhangqiquan <768617998@qq.com>
// +----------------------------------------------------------------------
declare (strict_types=1);

namespace app\api\admin;

use Simps\MQTT\Protocol\Types;
use Simps\MQTT\Protocol\V5;
use zhanshop\App;
use zhanshop\Log;

class MqttServEvent
{
    public function __construct()
    {
        App::phar()->import('extend/library/mqtt');
    }

    /**
     * @param \Swoole\Server $server
     * @param int $fd
     * @param int $reactorId
     * @param string $data
     * @return void
     */
    public function onReceive($server, $fd, $reactorId, $packData) :void{

        try {
            $data = \Simps\MQTT\Protocol\V5::unpack($packData);
        }catch (\Throwable $err){
            $server->close($fd);
            return;
        }

        try {
            switch ($data['type']){
                // 连接消息
                case Types::CONNECT:
                    if ($data['protocol_name'] != 'MQTT' || $data['protocol_level'] != 5) {
                        $server->close($fd);
                        return;
                    }
                    $this->connect($server, $fd);
                    break;
                // PING消息
                case Types::PINGREQ:
                    $this->ping($server, $fd);
                    break;
                case Types::DISCONNECT:
                    $this->disconnect($server, $fd);
                    break;
                // 推送广播
                case Types::PUBLISH:
                    foreach ($server->connections as $sub_fd) {
                        if ($sub_fd != $fd) {
                            $server->send(
                                $sub_fd,
                                V5::pack(
                                    [
                                        'type' => $data['type'],
                                        'topic' => $data['topic'],
                                        'message' => $data['message'],
                                        'dup' => $data['dup'],
                                        'qos' => $data['qos'],
                                        'retain' => $data['retain'],
                                        'message_id' => $data['message_id'] ?? 0,
                                    ]
                                )
                            );
                        }
                    }
                    if ($data['qos'] === 1) {
                        $server->send(
                            $fd,
                            V5::pack(
                                [
                                    'type' => Types::PUBACK,
                                    'message_id' => $data['message_id'] ?? 0,
                                    'message' => '123456',
                                ]
                            )
                        );
                    }
                    break;
                // 主题关注
                case Types::SUBSCRIBE:
                    $payload = [];
                    foreach ($data['topics'] as $k => $option) {
                        $qos = $option['qos'];
                        if (is_numeric($qos) && $qos < 3) {
                            $payload[] = $qos;
                        } else {
                            $payload[] = \Simps\MQTT\Hex\ReasonCode::QOS_NOT_SUPPORTED;
                        }
                    }
                    $server->send(
                        $fd,
                        V5::pack(
                            [
                                'type' => Types::SUBACK,
                                'message_id' => $data['message_id'] ?? 0,
                                'codes' => $payload,
                            ]
                        )
                    );
                    break;
                // 取消关注
                case Types::UNSUBSCRIBE:
                    $payload = [];
                    foreach ($data['topics'] as $qos) {
                        if (is_numeric($qos) && $qos < 3) {
                            $payload[] = $qos;
                        } else {
                            $payload[] = 0x80;
                        }
                    }
                    $server->send(
                        $fd,
                        V5::pack(
                            [
                                'type' => Types::UNSUBACK,
                                'message_id' => $data['message_id'] ?? 0,
                                'codes' => $payload,
                            ]
                        )
                    );
                    break;
            }
        }catch (\Throwable $err){
            Log::errorLog(5, $err->getMessage().PHP_EOL.$err->getFile().':'.$err->getLine());
            $server->close($fd);
        }
    }


    /**
     * 连接消息
     * @param $server
     * @param $fd
     * @return void
     */
    public function connect($server, $fd)
    {
        $server->send(
            $fd,
            V5::pack(
                [
                    'type' => Types::CONNACK,
                    'code' => 0,
                    'session_present' => 0,
                    'properties' => [
                        'maximum_packet_size' => 1048576,
                        'retain_available' => true,
                        'shared_subscription_available' => true,
                        'subscription_identifier_available' => true,
                        'topic_alias_maximum' => 65535, //0
                        'wildcard_subscription_available' => true,
                    ],
                ]
            )
        );
    }
    /**
     * ping消息
     * @return void
     */
    public function ping($server, $fd)
    {
        $server->send($fd, V5::pack(['type' => Types::PINGRESP]));
    }

    /**
     * 断开消息
     * @param $server
     * @param $fd
     * @return void
     */
    public function disconnect($server, $fd)
    {
        if ($server->exist($fd)) {
            $server->close($fd);
        }
    }

    public function publish()
    {

    }

}
